package deprecated;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.util.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import measures.DistanceSet;
import measures.Fuzzy_Match_Score;


public class MapReduce {

	private static String file;
	private static FileInputStream stream;
	private static ArrayList<String> testSentences;
	private static BufferedReader reader;

	
	
	/**
	 * MR implementation of mapper and reducer
	 * @author wangler
	 *
	 */
	public static class Map extends Mapper{
		
		//variables
		double fmTreshold = 0.7;
		DistanceSet currSent;
		static enum Counters { CURR_SENT }
		
		
		//map function
		public void map(Text key, Text value, Context context) throws IOException, InterruptedException{
			
			String source = testSentences.get((int)context.getCounter(Counters.CURR_SENT).getValue());
			
			String target = key.toString();	
			String tmtarget = value.toString();
			
			currSent = new DistanceSet(source, target);
			
			context.getCounter(Counters.CURR_SENT).increment(1);
			
			
			String keyout = source;
			if(currSent.getMaxFMS() < fmTreshold){	
				context.write(new Text(keyout), new Text(target + "\t" + tmtarget + "\t-1"));
			}else{
				context.write(new Text(keyout), new Text(target + "\t" + tmtarget + "\t" + Fuzzy_Match_Score.computeFMS(currSent)));
			}
			
		}
	}
	
	public static class Reduce extends Reducer{
		
		//variables
		double maxScore;
		String maxSent;
		
		public void reduce(Text key, Iterator<Text> values, Context context) throws IOException, InterruptedException {
			
			while(values.hasNext() && maxScore<1){
				
				Double currVal = Double.parseDouble(values.next().toString().split("\t")[2]); 
				String currPair = values.next().toString().split("\t")[0] + "\t" + values.next().toString().split("\t")[1];
				
				if(currVal > maxScore){
					maxScore = currVal;
					maxSent = currPair;
				}
			}	
		
			context.write(key + "\t" + maxSent, new Double(maxScore));
		}
		
		
	}
	
	
	public static void readFile() throws IOException{
		try {
			file = "/home/students/wangler/corpus.test.target.softpro.grp3";
			stream = new FileInputStream(file);
			reader = new BufferedReader(new InputStreamReader(stream));
			String line = reader.readLine();
			testSentences = new ArrayList<String>();
			testSentences.ensureCapacity(2000);
			
			while(line != null){
				
				testSentences.add(line);
				line = reader.readLine();
			}
				
		
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		reader.close();
			
		
		
		
		
		
	}
	
	
	/**
	 * @param args
	 * @throws IOException 
	 * @throws InterruptedException 
	 * @throws ClassNotFoundException 
	 */
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		
		readFile();
		
		Configuration conf = new Configuration();

		Job job = new Job(conf, "computeFMS");
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(DoubleWritable.class);
         
		
		job.setJarByClass(MapReduce.class);
		job.setMapperClass(deprecated.MapReduce.Map.class);
		job.setReducerClass(deprecated.MapReduce.Reduce.class);
		          
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		        
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		         
		job.waitForCompletion(true);
		

	}

}
